/*
 * Copyright (C) 2015-2020 Alibaba Group Holding Limited
 */

#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

/*
 *
 * Header file for aos kernel. Uncomment following if used.
 *
 */
/* #include "aos/kernel.h" */

/*
 *
 * Header file for yloop. Uncomment following if used.
 * e.g. Post got IP event.
 *
 */
/* #include "aos/yloop.h" */

/*
 *
 * Header file for ulog. Uncomment following if used.
 *
 */
#include "ulog/ulog.h"


/*
 *
 * Header file for at utility. Uncomment following if used.
 *
 */
#include "atparser.h"

#include "mal/mal.h"
#include "mal/hal_mal.h"

#include "atcmd_config_module.h"

#define MAL_HAL_TAG "mal_hal"

#define mal_debug(format, arg...)   LOGD(MAL_HAL_TAG, format, ##arg)
#define mal_info(format, arg...)    LOGI(MAL_HAL_TAG, format, ##arg)
#define mal_err(format, arg...)     LOGE(MAL_HAL_TAG, format, ##arg)

#define AT_EC20_MQTT_PREFIX            "+QMT"
#define AT_EC20_MQTT_CME_PREFIX        "+CME ERROR:"
#define AT_EC20_NET_CMD_PREFIX         "+C"

#define AT_EC20_MQTT_CONFIG            "+QMTCFG"
#define AT_EC20_MQTT_OPEN              "+QMTOPEN"
#define AT_EC20_MQTT_CONN              "+QMTCONN"
#define AT_EC20_MQTT_DISCON            "+QMTDISC"

#define AT_EC20_MQTT_SUB               "+QMTSUB"
#define AT_EC20_MQTT_UNSUB             "+QMTUNS"
#define AT_EC20_MQTT_PUB               "+QMTPUBEX"

#define AT_EC20_MQTT_RECV              "+QMTRECV"
#define AT_EC20_MQTT_STATE             "+QMTSTAT"
#define AT_EC20_MQTT_PING              "+QMTPING"

#define AT_CMD_PDP_DEACT               "+QIDEACT"
#define AT_CMD_PDP_CONFIG              "+QICSGP"
#define AT_CMD_PDP_ACT                 "+QIACT"


#define AT_EC20_MQTT_OK                "OK"
#define AT_EC20_MQTT_ERROR             "ERROR"
#define AT_EC20_PUB_PROMPT             "\r\n>"

#define AT_CMD_SIM_PIN_CHECK            "AT+CPIN?\r"
#define AT_CMD_SIGNAL_QUALITY_CHECK     "AT+CSQ\r"
#define AT_CMD_NETWORK_REG_CHECK        "AT+CREG?\r"
#define AT_CMD_ATTACH_CHECK             "AT+CGATT?\r"
#define AT_CMD_ECHO_OFF                 "ATE0\r"
#define AT_CMD_AT                       "AT\r"



#define AT_CMD_PREFIX_MAX_LEN          20
#define AT_CMD_RETRY_MAX               20

#if defined(MAL_CONFIG_MAX_TOPIC_LEN) && defined(MAL_CONFIG_MAX_MSG_LEN)
#define AT_MQTT_CMD_MAX_LEN             (MAL_CONFIG_MAX_TOPIC_LEN + MAL_CONFIG_MAX_MSG_LEN + 20)
#define AT_MQTT_RSP_MAX_LEN             (MAL_CONFIG_MAX_TOPIC_LEN + MAL_CONFIG_MAX_MSG_LEN + 20)
#else
#define AT_MQTT_CMD_MAX_LEN             400
#define AT_MQTT_RSP_MAX_LEN             400
#define MAL_CONFIG_MAX_TOPIC_LEN        128
#define MAL_CONFIG_MAX_MSG_LEN          256
#endif

#define AT_MQTT_WAIT_MAX_TIMEOUT            60*1000
#define AT_MQTT_WAIT_DEF_TIMEOUT            3*1000
#define AT_MQTT_WAIT_MIN_TIMEOUT            800

typedef enum {
   WAIT_NONE = 0,
   WAIT_SIMPLE_RSP,    /* OK or Error */
   WAIT_PIN_RSP,
   WAIT_REG_RSP,
   WAIT_SIGQ_RSP,
   WAIT_ATTCH_RSP,
   WAIT_OPEN_RSP,
   WAIT_CONN_RSP,
   WAIT_DISC_RSP,
   WAIT_SUB_RSP,
   WAIT_UNS_RSP,
   WAIT_PUB_PMT,
   WAIT_PUB_RSP,
} at_mqtt_send_state_t;

typedef enum {
   RSP_ERROR  = -1,
   RSP_OK     = 0,
   RSP_PROMPT = 3,
} at_mqtt_rsp_val_t;

enum {
    CONN_RESET_BY_PEER = 1,
    SEND_PING_FAIL  = 2,
    SEND_CONN_FAIL  = 3,
    RECV_CONNACK_FAIL = 4,
    SERVER_CLOSE_CONN = 5,
    CLIENT_CLOSE_CONN = 6,
    LINK_SERVER_UNAVAIABLE = 7,
    UNKNOWN = 8,
};

static volatile at_mqtt_send_state_t   m_atcmd_send_state = WAIT_NONE;
static volatile at_mqtt_rsp_val_t      m_ret_val = RSP_ERROR;

static volatile int        m_mqtt_connect_state = IOTX_MC_STATE_INVALID;
static volatile int        m_at_response_result = 0;
static iotx_mqtt_param_t  *m_pInitParams = NULL;
static void               *m_sem_response = NULL;

static volatile int       g_response_msg_number = 0;
static int                g_response_packetid = 0;
static int                g_response_status = 0;
static int                g_public_qos = 0;

static volatile unsigned int   m_msg_id = 0;

static char   at_max_len_cmd[AT_MQTT_CMD_MAX_LEN];
static char   at_max_recv_topic[MAL_CONFIG_MAX_TOPIC_LEN];
static char   at_max_recv_msg[MAL_CONFIG_MAX_MSG_LEN];

static uart_dev_t uart_dev;
int at_dev_fd = -1;
static mqtt_data_input_cb_t m_mqtt_data_input_cb = NULL;

static int at_ec20_mqtt_init(iotx_mqtt_param_t *pInitParams);
static int at_ec20_mqtt_config(void);
static int at_ec20_mqtt_conn(void);
static int at_ec20_mqtt_discon(void);
static int at_ec20_mqtt_unsub(const char *topic,
                              unsigned int *mqtt_packet_id,
                              int *mqtt_status);
static int at_ec20_mqtt_sub(const char *topic,
                     int qos,
                     unsigned int *mqtt_packet_id,
                     int *mqtt_status,
                     int timeout_ms);
static int at_ec20_mqtt_pub(const char *topic, 
                     int qos,
                     const char *message,
                     int msg_len);
static int at_ec20_mqtt_deinit(void);
static void at_ec20_mqtt_client_rsp_callback(void *arg, char *rspinfo, int rsplen);


/* To implement MQTT Init */
static int HAL_AT_MQTT_Init(iotx_mqtt_param_t *pInitParams)
{
    /* add your code here */
    return at_ec20_mqtt_init(pInitParams);;
}

/* To implement MQTT Connect */
static int HAL_AT_MQTT_Connect(char *proKey, char *devName, char *devSecret)
{
    (void)proKey;
    (void)devName;
    (void)devSecret;

    /* add your code here */
    return at_ec20_mqtt_conn();
}

/* To implement MQTT disconnect */
static int HAL_AT_MQTT_Disconnect(void)
{
     /* add your code here */
    return at_ec20_mqtt_discon();
}

/* To implement MQTT subscribe */
static int HAL_AT_MQTT_Subscribe(const char *topic, int qos, unsigned int *mqtt_packet_id, int *mqtt_status, int timeout_ms)
{
     /* add your code here */
    return at_ec20_mqtt_sub(topic, qos, mqtt_packet_id, mqtt_status, timeout_ms);
}

/* To implement MQTT unsubscribe */
static int HAL_AT_MQTT_Unsubscribe(const char *topic, unsigned int *mqtt_packet_id, int *mqtt_status)
{
     /* add your code here */
    return at_ec20_mqtt_unsub(topic, mqtt_packet_id, mqtt_status);
}

/* To implement MQTT publish */
static int HAL_AT_MQTT_Publish(const char *topic, int qos, const char *message, unsigned int msg_len)
{
     /* add your code here */
    return at_ec20_mqtt_pub(topic, qos, message, msg_len);
}

/* To implement MQTT state query */
static int HAL_AT_MQTT_State(void)
{
     /* add your code here */
    return (int) m_mqtt_connect_state;
}

/* To implement register MQTT data input callback */
static int HAL_MAL_RegisterMQTTDataInputCb(mqtt_data_input_cb_t cb)
{
    /* add your code here */
    if (cb) {
        m_mqtt_data_input_cb = cb;
        return 0;
    }

    return -1;
}

/* To implement MQTT deinit */
static int HAL_AT_MQTT_Deinit()
{
    /* add your code here */
    return at_ec20_mqtt_deinit();
}

/* To implement the driver configuration */
static int ec20_mal_add_dev(void* data)
{
    at_config_t at_config = { 0 };

    at_init();

    if(data != NULL)
    {
        mal_device_config_t* config = (mal_device_config_t *)data;
        uart_dev.port  = config->uart_dev.port;
        uart_dev.config.baud_rate    = config->uart_dev.config.baud_rate;
        uart_dev.config.data_width   = config->uart_dev.config.data_width;
        uart_dev.config.parity       = config->uart_dev.config.parity;
        uart_dev.config.stop_bits    = config->uart_dev.config.stop_bits;
        uart_dev.config.flow_control = config->uart_dev.config.flow_control;
        uart_dev.config.mode         = config->uart_dev.config.mode;
    }
    else
    {
        mal_err("Error: Uart dev is not configured! Please"
             " provide the config in sal_add_dev() call.\n");
        return -1;
    }

    /* configure and add one uart dev */
    at_config.type                             = AT_DEV_UART;
    at_config.port                             = uart_dev.port;
    at_config.dev_cfg                          = &uart_dev;
    at_config.send_wait_prompt                 = AT_SEND_DATA_WAIT_PROMPT;

    if ((at_dev_fd = at_add_dev(&at_config)) < 0) {
        mal_err("AT parser device add failed!\n");
        return -1;
    }

    at_register_callback(at_dev_fd, AT_EC20_MQTT_PREFIX,
                         NULL, NULL, 0,
                         at_ec20_mqtt_client_rsp_callback,
                         AT_EC20_MQTT_PREFIX);

    at_register_callback(at_dev_fd, AT_EC20_NET_CMD_PREFIX,
                         NULL, NULL, 0,
                         at_ec20_mqtt_client_rsp_callback,
                         AT_EC20_NET_CMD_PREFIX);
    
    at_register_callback(at_dev_fd, AT_EC20_MQTT_OK,
                         NULL, NULL, 0,
                         at_ec20_mqtt_client_rsp_callback,
                         AT_EC20_MQTT_OK);

    at_register_callback(at_dev_fd, AT_EC20_MQTT_ERROR,
                         NULL, NULL, 0,
                         at_ec20_mqtt_client_rsp_callback,
                         AT_EC20_MQTT_ERROR);

    at_register_callback(at_dev_fd, AT_EC20_PUB_PROMPT,
                         NULL, NULL, 0,
                         at_ec20_mqtt_client_rsp_callback,
                         AT_EC20_PUB_PROMPT);

   return 0;
}

int atcmd_wait_notify(int timeout_ms)
{
    return HAL_SemaphoreWait(m_sem_response, timeout_ms);
}

void atcmd_notify_sender(void)
{
    HAL_SemaphorePost(m_sem_response);
}

void atcmd_set_state(at_mqtt_send_state_t state)
{
    m_atcmd_send_state = state;
}

at_mqtt_send_state_t atcmd_get_state(void)
{
    return m_atcmd_send_state;
}

bool atcmd_send_state_ready(void)
{
    return (m_atcmd_send_state == WAIT_NONE);
}

at_mqtt_rsp_val_t atcmd_get_ret_val(void)
{
    return m_ret_val;
}

void atcmd_set_ret_val(at_mqtt_rsp_val_t ret)
{
    m_ret_val = ret;
}

typedef int (*at_data_check_cb_t)(char data);
int atcmd_num_check(char data)
{
    if ((data > '9' || data < '0') && data != '.') {
        return -1;
    }

    return 0;
}

int atcmd_info_get(char *buf, uint32_t buflen,
                               at_data_check_cb_t valuecheck)
{
    uint32_t i = 0;

    if (NULL == buf || 0 == buflen) {
        return -1;
    }

    do {
        if (at_read(at_dev_fd,&buf[i], 1) < 1) {
            mal_err("atcmd_read failed");
            return -1;
        }

        if (buf[i] == ',') {
            buf[i] = 0;
            break;
        } else if (buf[i] == '\r') {
            mal_debug(TAG, "delimiter found\n");
            buf[i] = 0;
            return 1;
        }

        if (i >= buflen) {
            mal_err("Too long length of data.reader is %s \r\n", buf);
            return -1;
        }
        if (NULL != valuecheck) {
            if (valuecheck(buf[i])) {
                mal_err("Invalid string!!!, reader is %s last char %d\r\n",
                     buf, buf[i]);
                return -1;
            }
        }
        i++;
    } while (1);

    return 0;
}


static int read_out_string(char *buffer, int size, const char *postfix)
{
    char single;
    int len = 0;

    /* read out prefix */
    do {
        if (at_read(at_dev_fd, &single, 1) < 0) {
            mal_err("at read failed!");
            return -1;
        }

        buffer[len++] = single;
        if (len >= size) {
            mal_err("Not found!");
            return -1;
        }

        if (len >= strlen(postfix) &&
            memcmp(postfix, buffer + len - strlen(postfix),
                   strlen(postfix)) == 0) {
            break;
        }
    } while(1);

    return len;
}

static void mqtt_config_rsp_process(void)
{
  /* not implement */
}

/* +QMTOPEN: 0,0 */
static void mqtt_open_rsp_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0, ret = -1;
    int client_id = -1, result = -1;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* client id */
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (client_id = atoi(prefix)) < 0) {
        mal_err("%s %d invalid client_id %s", __func__, __LINE__, prefix);
        return;
    }

     /* result */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 1 || (result = atoi(prefix)) < 0) {
        mal_err("%s %d invalid result %s", __func__, __LINE__, prefix);
        return;
    }

    if (atcmd_get_state() != WAIT_OPEN_RSP) {
        return;
    }

    mal_info("+QMTOPEN: %d,%d", client_id, result);

    if (0 == result) {
        atcmd_set_ret_val(RSP_OK);
    } else {
        atcmd_set_ret_val(RSP_ERROR);
    }

    atcmd_notify_sender();
}

/* +QMTCONN: 0,0,0 */
static void mqtt_conn_rsp_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0, ret = -1;
    int client_id = -1, result = -1, code = -1;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* client id */
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (client_id = atoi(prefix)) < 0) {
        mal_err("%s %d invalid client_id %s", __func__, __LINE__, prefix);
        return;
    }
    
     /* result */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret < 0 || (result = atoi(prefix)) < 0) {
        mal_err("%s %d invalid result %s", __func__, __LINE__, prefix);
        return;
    }

    /* code */
    if (ret == 0) {
        memset(prefix, 0, sizeof(prefix));
        ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
        if (ret != 1 || (code = atoi(prefix)) < 0) {
            mal_err("%s %d invalid code %s", __func__, __LINE__, prefix);
            return;
        }
    }

    if (atcmd_get_state() != WAIT_CONN_RSP) {
        return;
    }

    mal_info("+QMTCONN: %d,%d,%d", client_id, result, code);

    if (0 == result && 0 == code) {
        m_mqtt_connect_state = IOTX_MC_STATE_CONNECTED;
        atcmd_set_ret_val(RSP_OK);
    } else {
        atcmd_set_ret_val(RSP_ERROR);
    }

    atcmd_notify_sender();
}

/* +QMTDISC: 0,0 */
static void mqtt_disc_rsp_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0, ret = -1;
    int client_id = -1, result = -1;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* client id */
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (client_id = atoi(prefix)) < 0) {
        mal_err("%s %d invalid client_id %s", __func__, __LINE__, prefix);
        return;
    }
    
     /* result */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 1 || (result = atoi(prefix)) < 0) {
        mal_err("%s %d invalid result %s", __func__, __LINE__, prefix);
        return;
    }

    if (atcmd_get_state() != WAIT_DISC_RSP) {
        return;
    }

    mal_info("+QMTDISC: %d,%d", client_id, result);

    if (0 == result) {
        m_mqtt_connect_state = IOTX_MC_STATE_DISCONNECTED;
        atcmd_set_ret_val(RSP_OK);
    } else {
        atcmd_set_ret_val(RSP_ERROR);
    }

    atcmd_notify_sender();
}

/* +QMTSUB: 0,1,0,2 */
/* +QMTSUB: <client_idx>,<msgID>,<result>[,<value>] */
static void mqtt_sub_rsp_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0, ret = -1;
    int client_id = -1, msgID = -1, result = -1, value = -1;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* client id */
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (client_id = atoi(prefix)) < 0) {
        mal_err("%s %d invalid client_id %s", __func__, __LINE__, prefix);
        return;
    }

    /* msgID */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (msgID = atoi(prefix)) < 0) {
        mal_err("%s %d invalid msgID %s", __func__, __LINE__, prefix);
        return;
    }
    
     /* result */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret < 0 || (result = atoi(prefix)) < 0) {
        mal_err("%s %d invalid result %s", __func__, __LINE__, prefix);
        return;
    }

    /* value */
    if (ret == 0) {
        memset(prefix, 0, sizeof(prefix));
        ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
        if (ret != 1 || (value = atoi(prefix)) < 0) {
            mal_err("%s %d invalid value %s", __func__, __LINE__, prefix);
            return;
        }
    }

    if (atcmd_get_state() != WAIT_SUB_RSP) {
        return;
    }

    mal_info("+QMTSUB: %d,%d,%d,%d", client_id, msgID, result, value);

    if (0 == result) {
        atcmd_set_ret_val(msgID);
    } else {
        atcmd_set_ret_val(RSP_ERROR);
    }

    atcmd_notify_sender();
}

/* +QMTUNS: <client_idx>,<msgID>,<result>[,<value>] */
static void mqtt_unsub_rsp_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0, ret = -1;
    int client_id = -1, msgID = -1, result = -1, value = -1;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* client id */
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (client_id = atoi(prefix)) < 0) {
        mal_err("%s %d invalid client_id %s", __func__, __LINE__, prefix);
        return;
    }

    /* msgID */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (msgID = atoi(prefix)) < 0) {
        mal_err("%s %d invalid msgID %s", __func__, __LINE__, prefix);
        return;
    }
    
     /* result */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret < 0 || (result = atoi(prefix)) < 0) {
        mal_err("%s %d invalid result %s", __func__, __LINE__, prefix);
        return;
    }

    /* value */
    if (ret == 0) {
        memset(prefix, 0, sizeof(prefix));
        ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
        if (ret != 1 || (value = atoi(prefix)) < 0) {
            mal_err("%s %d invalid value %s", __func__, __LINE__, prefix);
            return;
        }
    }

    if (atcmd_get_state() != WAIT_UNS_RSP) {
        return;
    }

    mal_info("+QMTUNS: %d,%d,%d,%d", client_id, msgID, result, value);

    if (0 == result) {
        atcmd_set_ret_val(msgID);
    } else {
        atcmd_set_ret_val(RSP_ERROR);
    }

    atcmd_notify_sender();
}

/* +QMTPUBEX: 0,0,0 */
/* +QMTPUBEX: <client_idx>,<msgID>,<result>[,<value>] */
static void mqtt_pub_rsp_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0, ret = -1;
    int client_id = -1, msgID = -1, result = -1, value = -1;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* client id */
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (client_id = atoi(prefix)) < 0) {
        mal_err("%s %d invalid client_id %s", __func__, __LINE__, prefix);
        return;
    }

    /* msgID */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (msgID = atoi(prefix) < 0)) {
        mal_err("%s %d invalid msgID %s", __func__, __LINE__, prefix);
        return;
    }
    
     /* result */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret < 0 || (result = atoi(prefix)) < 0) {
        mal_err("%s %d invalid result %s", __func__, __LINE__, prefix);
        return;
    }

    /* value */
    if (ret == 0) {
        memset(prefix, 0, sizeof(prefix));
        ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
        if (ret != 1 || (value = atoi(prefix)) < 0) {
            mal_err("%s %d invalid value %s", __func__, __LINE__, prefix);
            return;
        }
    }

    if (atcmd_get_state() != WAIT_PUB_RSP) {
        return;
    }

    mal_info("+QMTPUBEX: %d,%d,%d,%d", client_id, msgID, result, value);

    if (0 == result) {
        atcmd_set_ret_val(msgID);
    } else {
        atcmd_set_ret_val(RSP_ERROR);
    }

    atcmd_notify_sender();
}

/* +QMTRECV: <client_idx>,<msg_id>,"<topic>",[<p ayload_len>,]"<payload>" */
/* +QMTRECV: 0,0,"topic/example",36,"This is the payload related to topic" */
static void mqtt_recv_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int client_id = -1, msgID = -1, len = -1, msglen = -1;
    int ret = -1, topic_len = 0;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* client id */
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (client_id = atoi(prefix)) < 0) {
        mal_err("%s %d invalid client_id %s", __func__, __LINE__, prefix);
        return;
    }

    /* msgID */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (msgID = atoi(prefix)) < 0) {
        mal_err("%s %d invalid msgID %s", __func__, __LINE__, prefix);
        return;
    }

    /* topic */
    at_read(at_dev_fd, &single, 1);
    memset(at_max_recv_topic, 0, sizeof(at_max_recv_topic));
    ret = atcmd_info_get(at_max_recv_topic, sizeof(at_max_recv_topic), NULL);
    if (ret != 0) {
        mal_err("%s %d get topic failed", __func__, __LINE__);
        return;
    }

    /* ignore \" */
    topic_len = strlen(at_max_recv_topic);
    if ( topic_len > 0 && at_max_recv_topic[topic_len - 1] == '\"') {
        at_max_recv_topic[topic_len - 1] = '\0';
        mal_info("recv topic %s", at_max_recv_topic);
    } else {
        mal_err("%s %d invalid topic %s", __func__, __LINE__, at_max_recv_topic);
        return;
    }

    /* msglen */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (msglen = atoi(prefix)) <= 0 ||
        msglen >= MAL_CONFIG_MAX_MSG_LEN) {
        mal_err("%s %d invalid msglen %s", __func__, __LINE__, prefix);
        return;
    }

    mal_info("recv msg len %d", msglen);
    memset(at_max_recv_msg, 0, sizeof(at_max_recv_msg));
    at_read(at_dev_fd, &single, 1);
    
    ret = at_read(at_dev_fd, at_max_recv_msg, msglen + 2);
    if (ret != msglen + 2) {
        mal_err("%s %d invalid msg", __func__, __LINE__);
        return;
    }
    at_max_recv_msg[msglen] = 0;

    if (m_mqtt_data_input_cb(at_max_recv_topic, strlen(at_max_recv_topic), at_max_recv_msg, msglen) != 0) {
        mal_err("hand data to uplayer fail!\n");
    }
}

/* +QMTSTAT: <client_idx>,<err_code> */
static void mqtt_state_process(void)
{
    char prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0;
    int client_id = -1, code = -1, ret = -1;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* client id */
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (client_id = atoi(prefix)) < 0) {
        mal_err("%s %d invalid client_id %s", __func__, __LINE__, prefix);
        return;
    }

    /* msgID */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 1 || (code = atoi(prefix)) < 0) {
        mal_err("%s %d invalid code %s", __func__, __LINE__, prefix);
        return;
    }

    mal_info("+QMTSTAT: %d,%d", client_id, code);

    /* TODO: act on code */
    m_mqtt_connect_state = IOTX_MC_STATE_DISCONNECTED;
}

static void mqtt_ping_process(void)
{
   /* TODO: act on ping */
}

static void net_pin_check_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0;

    /* read out prefix */
    len = read_out_string(prefix, AT_CMD_PREFIX_MAX_LEN, "OK\r\n");
    if (len < 0) {
        return;
    }
    
    if (atcmd_get_state() != WAIT_PIN_RSP) {
        return;
    }

    if (strstr(prefix, "READY") != NULL) {
        atcmd_set_ret_val(RSP_OK);
    } else {
        atcmd_set_ret_val(RSP_ERROR);
    }

    atcmd_notify_sender();
}

static void net_sig_quality_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0, ret = -1;
    int rssi, ber;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* rssi */
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (rssi = atoi(prefix)) < 0) {
        mal_err("%s %d invalid rssi %s", __func__, __LINE__, prefix);
        return;
    }

    /* ber */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 1 || (ber = atoi(prefix)) < 0) {
        mal_err("%s %d invalid ber %s", __func__, __LINE__, prefix);
        return;
    }

    read_out_string(prefix, sizeof(prefix), "OK\r\n");

    if (atcmd_get_state() != WAIT_SIGQ_RSP) {
        return;
    }

    mal_info("Sig quality %d %d", rssi, ber);

    if (rssi > 0 && rssi < 99) {
        atcmd_set_ret_val(RSP_OK);
    } else {
        atcmd_set_ret_val(RSP_ERROR);
    }

    atcmd_notify_sender();
}

/*AT+CEREG?
+CEREG: 0,1
*/
static void net_register_process()
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0, ret = -1;
    int n = 0, stat = 0;

    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* n */
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 0 || (n = atoi(prefix)) < 0) {
        mal_err("%s %d invalid rssi %s", __func__, __LINE__, prefix);
        return;
    }

    /* stat */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 1 || (stat = atoi(prefix)) < 0) {
        mal_err("%s %d invalid ber %s", __func__, __LINE__, prefix);
        return;
    }

    read_out_string(prefix, sizeof(prefix), "OK\r\n");

    if (atcmd_get_state() != WAIT_REG_RSP) {
        return;
    }

    mal_info("Register state %d %d", n, stat);

    if (1 == stat) {
        atcmd_set_ret_val(RSP_OK);
    } else {
        atcmd_set_ret_val(RSP_ERROR);
    }

    atcmd_notify_sender();
}

static void net_attach_process()
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0, ret = -1;
    int attach = 0;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* attach */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 1 || (attach = atoi(prefix)) < 0) {
        mal_err("%s %d invalid attach %s", __func__, __LINE__, prefix);
        return;
    }
    
    read_out_string(prefix, sizeof(prefix), "OK\r\n");

    if (atcmd_get_state() != WAIT_ATTCH_RSP) {
        return;
    }

    mal_info("Attach state %d", attach);

    if (1 == attach) {
        atcmd_set_ret_val(RSP_OK);
    } else {
        atcmd_set_ret_val(RSP_ERROR);
    }

    atcmd_notify_sender();
}

static void cme_error_process()
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0, ret = -1;
    int error = 0;
    char single;

    /* eat space */
    at_read(at_dev_fd, &single, 1);

    /* error */
    memset(prefix, 0, sizeof(prefix));
    ret = atcmd_info_get(prefix, sizeof(prefix), atcmd_num_check);
    if (ret != 1 || (error = atoi(prefix)) < 0) {
        mal_err("%s %d invalid error %s", __func__, __LINE__, prefix);
        return;
    }
    
    if (atcmd_get_state() == WAIT_NONE) {
        return;
    }

    mal_info("CME Error: %d, current state %d", error, atcmd_get_state());

    atcmd_set_ret_val(RSP_ERROR);

    atcmd_notify_sender();
}

static void at_net_cmd_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    int len = 0;

    /* read out prefix */
    len = read_out_string(prefix, AT_CMD_PREFIX_MAX_LEN, ":");
    if (len < 0) {
        return;
    }

    prefix[len - 1] = '\0';

    if (memcmp(prefix, "PIN", strlen(prefix)) == 0) {
        net_pin_check_process(); 
    } else if (memcmp(prefix, "SQ", strlen(prefix)) == 0) {
        net_sig_quality_process();
    } else if (memcmp(prefix, "REG", strlen(prefix)) == 0) {
        net_register_process();
    } else if (memcmp(prefix, "GATT", strlen(prefix)) == 0) {
        net_attach_process();
    } else if (memcmp(prefix, "ME ERROR", strlen(prefix)) == 0) {
        cme_error_process();
    } else {
        mal_err("Unknown preifx %s!", prefix);
    }
}

static void at_rsp_process(at_mqtt_rsp_val_t rsp_val)
{
    if (atcmd_get_state() != WAIT_SIMPLE_RSP &&
        atcmd_get_state() != WAIT_PUB_PMT) {
        mal_debug("%s state %d val %d", __func__, atcmd_get_state(), rsp_val);
        return;
    }

    atcmd_set_ret_val(rsp_val);

    atcmd_notify_sender();
}

static void at_mqtt_process(void)
{
    char  prefix[AT_CMD_PREFIX_MAX_LEN] = { 0 };
    char  single;
    int i = 0;

    /* read out prefix */
    do {
        at_read(at_dev_fd, &single, 1);
        prefix[i++] = single;
        if (i >= AT_CMD_PREFIX_MAX_LEN) {
            mal_err("Prefix too long!");
            return;
        }
    } while(single != ':');

    prefix[i - 1] = '\0';

    mal_debug("Recv Prefix %s", prefix);

    if (memcmp(prefix, "RECV", strlen(prefix)) == 0) {
        mqtt_recv_process(); 
    } else if (memcmp(prefix, "PUBEX", strlen(prefix)) == 0) {
        mqtt_pub_rsp_process();
    } else if (memcmp(prefix, "CONN", strlen(prefix)) == 0) {
        mqtt_conn_rsp_process();
    } else if (memcmp(prefix, "DISC", strlen(prefix)) == 0) {
        mqtt_disc_rsp_process();
    } else if (memcmp(prefix, "SUB", strlen(prefix)) == 0) {
        mqtt_sub_rsp_process();
    } else if (memcmp(prefix, "UNS", strlen(prefix)) == 0) {
        mqtt_unsub_rsp_process();
    } else if (memcmp(prefix, "OPEN", strlen(prefix)) == 0) {
        mqtt_open_rsp_process();
    } else if (memcmp(prefix, "CFG", strlen(prefix)) == 0) {
       mqtt_config_rsp_process();
    } else if (memcmp(prefix, "STAT", strlen(prefix)) == 0) {
        mqtt_state_process();
    } else if (memcmp(prefix, "PING", strlen(prefix)) == 0) {
        mqtt_ping_process();
    } else {
        mal_err("Unknown preifx %s!", prefix);
    }
}

static void at_ec20_mqtt_client_rsp_callback(void *arg, char *rspinfo, int rsplen)
{
    char *input = (char *)arg;

    if (NULL == input) {
        mal_err("invalid input of rsp callback");
        return;
    }

    if (0 == memcmp(input, AT_EC20_MQTT_PREFIX,
                    strlen(AT_EC20_MQTT_PREFIX))) {
        at_mqtt_process();
    } else if (0 == memcmp(input, AT_EC20_NET_CMD_PREFIX,
                    strlen(AT_EC20_NET_CMD_PREFIX))) {
        at_net_cmd_process();
    } else if (0 == memcmp(input, AT_EC20_MQTT_OK,
                    strlen(AT_EC20_MQTT_OK))) {
        at_rsp_process(RSP_OK);
    } else if (0 == memcmp(input, AT_EC20_MQTT_ERROR,
                    strlen(AT_EC20_MQTT_ERROR))) {
        at_rsp_process(RSP_ERROR);
    } else if (0 == memcmp(input, AT_EC20_PUB_PROMPT,
                    strlen(AT_EC20_PUB_PROMPT))) {
        at_rsp_process(RSP_PROMPT);
    } else {
        mal_err("invalid input of rsp callback");
        return;
    }

    return;
}

static int atcmd_send_wait_ret_val(const char *data, int datalen,
                                  at_mqtt_send_state_t state, int timeout_ms) 
{
    int ret = 0;

    if (!atcmd_send_state_ready()) {
        mal_err("Not in ready state\r\n");
        return -1;
    }

    atcmd_set_state(state);

    ret = at_send_no_reply(at_dev_fd, data, datalen, false);
    if (ret < 0) {
        atcmd_set_state(WAIT_NONE);
        mal_err("at send failed");
        return -1;
    }

    if (timeout_ms > 0) { 
        if (0 == atcmd_wait_notify(timeout_ms)) {
            ret = atcmd_get_ret_val();
        } else {
            ret = -1;
        }
    }
    
    atcmd_set_state(WAIT_NONE);

    return ret;
}

static int at_ec20_echo_off(void)
{      
    return atcmd_send_wait_ret_val(AT_CMD_ECHO_OFF, strlen(AT_CMD_ECHO_OFF),
                                   WAIT_SIMPLE_RSP, AT_MQTT_WAIT_DEF_TIMEOUT); 
}

static int at_ec20_module_ready(void)
{      
    return atcmd_send_wait_ret_val(AT_CMD_AT, strlen(AT_CMD_AT),
                                   WAIT_SIMPLE_RSP, AT_MQTT_WAIT_DEF_TIMEOUT); 
}


static int at_ec20_lte_status_check(void)
{
    int retry = 0;
    int ret;

    do {
        retry++;
        /*sim card status check*/
        ret = atcmd_send_wait_ret_val(AT_CMD_SIM_PIN_CHECK, strlen(AT_CMD_SIM_PIN_CHECK),
                                      WAIT_PIN_RSP, AT_MQTT_WAIT_DEF_TIMEOUT);
    } while (ret != 0 && retry < AT_CMD_RETRY_MAX);

    /* Signal quaility check */
    ret = atcmd_send_wait_ret_val(AT_CMD_SIGNAL_QUALITY_CHECK, strlen(AT_CMD_SIGNAL_QUALITY_CHECK),
                                  WAIT_SIGQ_RSP, AT_MQTT_WAIT_DEF_TIMEOUT);
    if (ret != 0) {
        mal_err("Signal quaility check failed\r\n");
        return -1;
    }

    /* network registration check*/
    ret = atcmd_send_wait_ret_val(AT_CMD_NETWORK_REG_CHECK, strlen(AT_CMD_NETWORK_REG_CHECK),
                                  WAIT_REG_RSP, AT_MQTT_WAIT_DEF_TIMEOUT);
    if (ret != 0) {
        mal_err("Network registration check failed\r\n");
        return -1;
    }

    /* PS attach check*/
    ret = atcmd_send_wait_ret_val(AT_CMD_ATTACH_CHECK, strlen(AT_CMD_ATTACH_CHECK),
                                  WAIT_ATTCH_RSP, AT_MQTT_WAIT_DEF_TIMEOUT);
    if (ret != 0) {
        mal_err("Attchement check failed\r\n");
        return -1;
    }

    return 0;
}

static int at_ec20_prepare_network(void)
{
    int ret = 0;
    int retry = 0;

    /* wait module ready */
    do {
       retry++;
       ret = at_ec20_module_ready();
       if (ret != 0) mal_err("%s %d retry ...\r\n", __func__, __LINE__);
    } while(ret != 0 && retry < AT_CMD_RETRY_MAX);
    
    /* turn off echo */
    do {
       retry++;
       ret = at_ec20_echo_off();
       if (ret != 0) mal_err("%s %d retry ...\r\n", __func__, __LINE__);
    } while(ret != 0 && retry < AT_CMD_RETRY_MAX);

    if (ret != 0)
       return -1;

    /* lte status */
    do {
       retry++;
       ret = at_ec20_lte_status_check();
       if (ret != 0) mal_err("%s %d retry ...\r\n", __func__, __LINE__);
    } while(ret != 0 && retry < AT_CMD_RETRY_MAX);

    return ret;
}

static int at_ec20_pdp_config(void)
{
    int ret = 0;
    int contextid = 1;
    int context_type = 1;  /*IPV4*/
    int auth_type = 0;     /*none*/
    
#define APN_TYPE    "\"CMNET\""
#define APN_USERNAME "\"\""
#define APN_PASSWORKD "\"\""

    memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));
    HAL_Snprintf(at_max_len_cmd,
                sizeof(at_max_len_cmd),
                "AT%s=%d\r", AT_CMD_PDP_DEACT, contextid);
    if (0 != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                         WAIT_SIMPLE_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
        mal_err("%s cmd fail", at_max_len_cmd);
        return -1;
    }

    memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));
    HAL_Snprintf(at_max_len_cmd,
                sizeof(at_max_len_cmd),
                "AT%s=%d,%d,%s,%s,%s,%d\r", AT_CMD_PDP_CONFIG, 
                contextid, context_type, APN_TYPE, APN_USERNAME,
                APN_PASSWORKD, auth_type);
    if (0 != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                         WAIT_SIMPLE_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
        mal_err("%s cmd fail", at_max_len_cmd);
        return -1;
    }
                                         
    memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));
    HAL_Snprintf(at_max_len_cmd,
                sizeof(at_max_len_cmd),
                "AT%s=%d\r", AT_CMD_PDP_ACT, contextid);
    if (0 != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                         WAIT_SIMPLE_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
        mal_err("%s cmd fail", at_max_len_cmd);
        return -1;
    }

    return 0;
}

static int at_ec20_mqtt_config(void)
{
    int   ret;
    int   client_id = 0;   /* only one instance by defaul */

    if (m_pInitParams == NULL) {
        mal_err("init params is not configured.");
        return -1;
    }

    /* config recv mode */ 
    {
        int   msg_recv_mode = 0;
        int   msg_len_enable = 1;
        memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));
        HAL_Snprintf(at_max_len_cmd,
                    sizeof(at_max_len_cmd),
                    "AT%s=\"recv/mode\",%d,%d,%d\r",
                    AT_EC20_MQTT_CONFIG, client_id, msg_recv_mode, msg_len_enable);

        if (0 != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                         WAIT_SIMPLE_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
            mal_err("config recv mode command fail");
            return -1;
        }
    }

    /* config ssl mode */
    {
        int  ssl_enable = 1;
        int  ssl_ctx_id = 0;
        memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));
        HAL_Snprintf(at_max_len_cmd,
                    sizeof(at_max_len_cmd),
                    "AT%s=\"SSL\",%d,%d,%d\r",
                    AT_EC20_MQTT_CONFIG, client_id, ssl_enable, ssl_ctx_id);

        if (0 != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                         WAIT_SIMPLE_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
            mal_err("config ssl at command fail");
            return -1;
        }
    }
    
    /* clean mqtt session */
    {
        int  clean_session = 1;
        memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));
        HAL_Snprintf(at_max_len_cmd,
                    sizeof(at_max_len_cmd),
                    "AT%s=\"session\",%d,%d\r",
                    AT_EC20_MQTT_CONFIG, client_id, clean_session);

        if (0 != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                         WAIT_SIMPLE_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
            mal_err("config session at command fail");
            return -1;
        }
    }

    /* send packet timeout */
    {
        int retry_times = 3;
        int timeout_notify = 0;
        memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));
        HAL_Snprintf(at_max_len_cmd,
                     sizeof(at_max_len_cmd),
                     "AT%s=\"timeout\",%d,%d,%d,%d\r",
                     AT_EC20_MQTT_CONFIG, client_id, m_pInitParams->request_timeout_ms/1000,
                     retry_times, timeout_notify);

        if (0 != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                         WAIT_SIMPLE_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
            mal_err("config timeout at command fail");
            return -1;
        }
    }

    return 0;
}

static int at_ec20_mqtt_conn(void)
{
    int   client_id = 0;   /* only one instance by default */

    if (m_pInitParams == NULL) {
        mal_err("init params is not configured.");
        return -1;
    }

    /* open network for mqtt client */ 
    {
        memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));
        HAL_Snprintf(at_max_len_cmd,
                    sizeof(at_max_len_cmd),
                    "AT%s=%d,\"%s\",%d\r",
                    AT_EC20_MQTT_OPEN, client_id, m_pInitParams->host, m_pInitParams->port);

        if (0 != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                         WAIT_OPEN_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
            mal_err("open at command fail");
            return -1;
        }
    }

    /* connect to server */
    {
        memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));
        HAL_Snprintf(at_max_len_cmd,
                    sizeof(at_max_len_cmd),
                    "AT%s=%d,\"%s\",\"%s\",\"%s\"\r",
                    AT_EC20_MQTT_CONN, client_id, m_pInitParams->client_id, m_pInitParams->username, m_pInitParams->password);

        if (0 != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                         WAIT_CONN_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
            mal_err("conn %s  at command fail", at_max_len_cmd);
            return -1;
        }
    }

    return 0;
}

static int at_ec20_mqtt_sub(const char *topic,
                     int qos,
                     unsigned int *mqtt_packet_id,
                     int *mqtt_status,
                     int timeout_ms)
{
    int   client_id = 0;   /* only one instance by default */

    if ((topic == NULL) || (mqtt_packet_id == NULL) || (mqtt_status == NULL)) {
        mal_err("subscribe param should not be NULL");
        return -1;
    }

    memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));

    HAL_Snprintf(at_max_len_cmd,
                 sizeof(at_max_len_cmd) - 1,
                 "AT%s=%d,%d,\"%s\",%d\r",
                 AT_EC20_MQTT_SUB, client_id,
                 m_msg_id, topic, qos);
   
    if (m_msg_id != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                            WAIT_SUB_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
        mal_err("conn at command fail");
        m_msg_id++;
        return -1;
    }

    m_msg_id++;

    return 0;
}

static int at_ec20_mqtt_unsub(const char *topic,
                              unsigned int *mqtt_packet_id,
                              int *mqtt_status)
{
    int   client_id = 0;   /* only one instance by default */

    if ((topic == NULL) || (mqtt_packet_id == NULL) || (mqtt_status == NULL)) {
        mal_err("unsubscribe param should not be NULL");
        return -1;
    }

    memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));

    HAL_Snprintf(at_max_len_cmd,
                 sizeof(at_max_len_cmd) - 1,
                 "AT%s=%d,%d,\"%s\"\r",
                 AT_EC20_MQTT_UNSUB, client_id, 
                 m_msg_id, topic);

    if (m_msg_id != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                     WAIT_UNS_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
        mal_err("unsub at command fail");
        m_msg_id++;
        return -1;
    }

    m_msg_id++;

    return 0;
}

static int at_ec20_mqtt_pub(const char *topic, int qos, const char *message, int msg_len)
{
    int   client_id = 0;   /* only one instance by default */
    int   retain = 0;
    int   msg_id = 0;

    if ((topic == NULL) || (message == NULL)) {
        mal_err("publish param should not be NULL");
        return -1;
    }

    if (qos > 0) {
        msg_id = m_msg_id;
    }

    HAL_Snprintf(at_max_len_cmd,
                 sizeof(at_max_len_cmd) - 1,
                 "AT%s=%d,%d,%d,%d,\"%s\",%d\r",
                 AT_EC20_MQTT_PUB,
                 client_id, msg_id,
                 qos, retain, topic, msg_len);

    if (RSP_PROMPT != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                              WAIT_PUB_PMT, AT_MQTT_WAIT_DEF_TIMEOUT)) {
        mal_err("wait prompt fail");
        m_msg_id++;
        return -1;
    }

    if (msg_id != atcmd_send_wait_ret_val(message, msg_len,
                                          WAIT_PUB_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
        mal_err("pub at cmd fail");
        m_msg_id++;
        return -1;
    }

    m_msg_id++;

    return 0;
}

static int at_ec20_mqtt_discon(void)
{
    int   client_id = 0;   /* only one instance by default */

    memset(at_max_len_cmd, 0, sizeof(at_max_len_cmd));

    HAL_Snprintf(at_max_len_cmd,
                 sizeof(at_max_len_cmd) - 1,
                 "AT%s=%d\r",
                 AT_EC20_MQTT_DISCON, client_id);

    if (0 != atcmd_send_wait_ret_val(at_max_len_cmd, strlen(at_max_len_cmd),
                                     WAIT_DISC_RSP, AT_MQTT_WAIT_DEF_TIMEOUT)) {
        mal_err("discon at command fail");
        return -1;
    }

    return 0;
}

static int at_ec20_mqtt_init(iotx_mqtt_param_t *pInitParams)
{
    int ret = 0;

    if (NULL == pInitParams)
        return -1;

    if (NULL == (m_sem_response = HAL_SemaphoreCreate())) {
        mal_err("Create sem failed");
        return -1;
    }

    m_mqtt_connect_state = IOTX_MC_STATE_INVALID;
    m_pInitParams = pInitParams;

    /* prepare lte network */
    ret = at_ec20_prepare_network();
    if (ret != 0) {
        mal_err("prepare network failed");
        return -1;
    }
    /*pdp config */
    ret = at_ec20_pdp_config();
    if (ret != 0) {
        mal_err("pdp config failed");
        return -1;
    }

    /* config mqtt */
    ret = at_ec20_mqtt_config();
    if (ret != 0) {
        mal_err("mqtt config failed");
        return -1;
    }
    
    //at_ec20_mqtt_discon();

    return 0;
}

static int at_ec20_mqtt_deinit(void)
{
    HAL_SemaphoreDestroy(m_sem_response);
    m_sem_response = NULL;
    m_mqtt_connect_state = IOTX_MC_STATE_INVALID;
    m_msg_id = 0;

    return 0;
}

/* Don't modify */
static mal_op_t mal_op = {
    .next = NULL,
    .version = "1.0.0",
    .name = "ec20",
    .add_dev = ec20_mal_add_dev,
    .init = HAL_AT_MQTT_Init,
    .connect = HAL_AT_MQTT_Connect,
    .subscribe = HAL_AT_MQTT_Subscribe,
    .unsubscribe = HAL_AT_MQTT_Unsubscribe,
    .publish = HAL_AT_MQTT_Publish,
    .disconn = HAL_AT_MQTT_Disconnect,
    .deinit = HAL_AT_MQTT_Deinit,
    .conn_state = HAL_AT_MQTT_State,
    .register_mqtt_data_input_cb = HAL_MAL_RegisterMQTTDataInputCb,
};

/* Generated by aos tools, do NOT modify! */
#ifndef __MAL_DEV_NAME
#error "__MAL_DEV_NAME C macro not defined, please define it in the MAL device driver."
#endif

int __MAL_DEV_NAME_mal_device_init(void)
{
    return mal_module_register(&mal_op);
}

